package com.at.bigdata.spark.core.req

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 *
 * @author cdhuangchao3
 * @date 2023/5/19 10:02 PM
 */
object Spark02_Req1_HotCategoryTop10Analysis_union {

  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("operator")
    val sc = new SparkContext(sparkConf)

    // Q1: rdd重复使用
    // Q2: 性能可能较低

    // 1、读取原始数据
    val actionRDD = sc.textFile("datas/user_visit_action.txt")
    actionRDD.cache()

    // 2、统计品类的点击数量：（品类ID，点击数量）
    val clickActionRDD = actionRDD.filter {
      action => {
        val datas = action.split("_")
        datas(6) != "-1"
      }
    }
    val clickCountRDD = clickActionRDD.map(
      action => {
        val datas = action.split("_")
        (datas(6), 1)
      }
    ).reduceByKey(_ + _)

    // 3、统计品类的下单数量：（品类ID，下单数量）
    val orderActionRDD = actionRDD.filter(
      action => {
        val datas = action.split("_")
        datas(8) != "null"
      }
    )
    val orderCountRDD = orderActionRDD.flatMap(
      action => {
        val datas = action.split("_")
        val cids = datas(8).split(",")
        cids.map(d => (d, 1))
      }
    ).reduceByKey(_ + _)

    // 4、统计品类的支付数量：（品类ID，支付数量）
    val payActionRDD = actionRDD.filter(
      action => {
        val datas = action.split("_")
        datas(10) != "null"
      }
    )
    val payCountRDD = payActionRDD.flatMap(
      action => {
        val datas = action.split("_")
        datas(10).split(",").map(d => (d, 1))
      }
    ).reduceByKey(_ + _)

    // 5、将品类排序，并且取前十名
    // （品类ID，（点击数量，下单数量，支付数量））
    // cogroup = connect + group
    // 有可能存在shuffle
    val rdd1 = clickCountRDD.map {
      case (cid, cnt) => {
        (cid, (cnt, 0, 0))
      }
    }
    val rdd2 = orderCountRDD.map {
      case (cid, cnt) => {
        (cid, (0, cnt, 0))
      }
    }
    val rdd3 = payCountRDD.map {
      case (cid, cnt) => {
        (cid, (0, 0, cnt))
      }
    }
    // 将3个数据源合并在一起，统一进行聚合计算
    val sourceRDD = rdd1.union(rdd2).union(rdd2)

    val analysisRDD = sourceRDD.reduceByKey(
      (t1, t2) => {
        (t1._1 + t2._1, t1._2 + t2._2, t1._3 + t2._3)
      }
    )
    val resultRDD = analysisRDD.sortBy(_._2, false).take(10)
    // 6、将结果采集到控制台打印出来
    resultRDD.foreach(println)
    sc.stop()
  }

}
